package org.ykhl.sdk.appserver.xiacheng;

import org.ykhl.mq.commons.BaseMessage;
import org.ykhl.mq.commons.ConnectionException;
import org.ykhl.mq.commons.Constants;
import org.ykhl.mq.commons.receiver.MessageConsumeStatus;
import org.ykhl.mq.commons.receiver.MessageListener;
import org.ykhl.sdk.appserver.MessageQueueFactoryKafkaImpl;
import org.ykhl.sdk.appserver.MessageQueueKafkaImpl;

import java.util.Scanner;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * Created by Zhangkefei on 2016/2/26.
 */
public class MessageQueueTestManual {

    public static void main(String[] args) throws ConnectionException, InterruptedException {
        final String authId = (args.length >= 1) ? args[0] : "2ioQwVLmV1CGIN2uL8VpIp";
//        final String authId = (args.length >= 1) ? args[0] : "22iQZMV093Ja16g1q8ZBwS";
        final String server = (args.length >= 2) ? args[1] : "dev";
        System.out.println(String.format("using authId: %s, server type: %s", authId, server));

        MessageQueueKafkaImpl messageQueue = MessageQueueFactoryKafkaImpl.newInstance(authId);
        if(! server.equals("prod"))
            messageQueue.setServerAddress(Constants.KAFKA_SERVER_LIST_DEV);

        messageQueue.setListener(new MessageListener() {

            @Override
            public MessageConsumeStatus consume(BaseMessage msg) {
                //业务逻辑处理部分
                System.out.println(msg);
                return MessageConsumeStatus.CONSUME_SUCCESS;
            }
        });
        messageQueue.start();

        Scanner scanner = new Scanner(System.in);
        while(!scanner.nextLine().equals("quit")){
            System.out.print("input fid: ");
            String fid = scanner.nextLine();
            System.out.println("input request content:");
            StringBuilder sb = new StringBuilder();
            String line;
            while (! (line = scanner.nextLine()).equals("q")) {
                sb.append(line);
            }

            BlockingQueue<BaseMessage> queue = messageQueue.sendRequest(
                    fid, "5.0", sb.toString(), "devSign", "1456718064634");
            BaseMessage msg = queue.poll(60, TimeUnit.SECONDS);
            if (msg != null){
                System.out.println("success!!!");
                System.out.println(msg);
            }
            else
                System.out.println("timeout");
        }
        scanner.close();

        //调用stop()方法停止接收消息
        messageQueue.stop();
    }

}
